package cn.doitedu.rtdw.dataetl;

import cn.doitedu.rtdw.udfs.GpsToGeohashUdf;
import cn.doitedu.rtdw.udfs.MapToJsonStr;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class JOB01_行为日志公共维度退维 {

    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(5000);
        //env.getCheckpointConfig().setCheckpointStorage("hdfs://doitedu:8020/rtdw_job1_ckpt");
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        // 建表，映射kafka中的原始行为日志topic
        // {"release_channel":"360应用市场","device_type":"mi6","session_id":"s01","lat":39.89313025958401,"lng":116.42883455189471,"username":"tiger","eventId":"app_launch","timestamp":1670583693000,"properties":{"url":"/content/article/2354.html?a=3","itemId":"item002"}}
        tenv.executeSql(
                " create table action_log_kafka(                          "+
                        "     release_channel string,                             "+
                        "     device_type string,                                 "+
                        "     session_id  string,                                 "+
                        "     lat         double,                                 "+
                        "     lng         double,                                 "+
                        "     username    string,                                 "+
                        "     eventId     string,                                 "+
                        "     `timestamp`   bigint,                                 "+
                        "     properties  map<string,string>,                     "+
                        // 表达式字段
                        "     pt AS  proctime()                                   "+  // 代表处理时间语义的时间
                        " ) with (                                                "+
                        "     'connector' = 'kafka',                              "+
                        "     'topic' = 'ods-events',                             "+
                        "     'properties.bootstrap.servers' = 'doitedu:9092',    "+
                        "     'properties.group.id' = 'goo1',                     "+
                        "     'scan.startup.mode' = 'latest-offset',              "+
                        "     'value.format'='json',                              "+
                        "     'value.json.fail-on-missing-field'='false',         "+
                        "     'value.fields-include' = 'EXCEPT_KEY'               "+
                        " )                                                       "
        );


        // 建表，映射hbase中的维表：用户注册信息表
        tenv.executeSql(
                " create table user_hbase(                 "+
                        " 	username string,                      "+
                        "   f  row<                                " +
                        "          id  bigint                      "+
                        " 	       ,member_level_id bigint         "+
                        "          ,password string                "+
                        "          ,nickname string                "+
                        "          ,phone string                   "+
                        "          ,status int                     "+
                        "          ,create_time timestamp(3)       "+
                        "          ,icon string                    "+
                        "          ,gender int                     "+
                        "          ,birthday date                  "+
                        "          ,city string                    "+
                        "          ,job string                     "+
                        "          ,personalized_signature string  "+
                        "          ,source_type int                "+
                        "          ,integration int                "+
                        "          ,growth int                     "+
                        "          ,luckey_count int               "+
                        "          ,history_integration int        "+
                        "          ,modify_time timestamp(3)       "+
                        " 	>                                      "+
                        " ) WITH (                                 "+
                        "  'connector' = 'hbase-2.2',              "+
                        "  'table-name' = 'doit42_user_info',      "+
                        "  'zookeeper.quorum' = 'doitedu:2181'     "+
                        " )                                        "
        );



        // 建表，映射hbase中的维表：地理位置维表
        tenv.executeSql(
                " create table area_hbase(                            "+
                        "     geohash string,                                  "+
                        " 	  f row<province string,city string,region string>   "+
                        " ) with (                                             "+
                        "     'connector' = 'hbase-2.2',                       "+
                        "     'table-name' = 'doit42_area',                    "+
                        "     'zookeeper.quorum' = 'doitedu:2181',             "+
                        "     'lookup.cache' = 'PARTIAL',                      "+
                        "     'lookup.partial-cache.max-rows' = '100000'       "+
                        " )                                                    "
        );



        // 建表，映射hbase中的维表：页面信息维表
        tenv.executeSql(
                " create table page_hbase(                            "+
                        "     url_prefix string,                                 "+
                        " 	  f row<page_type string,page_service string,page_lanmu string,page_pindao string>   "+
                        " ) with (                                             "+
                        "     'connector' = 'hbase-2.2',                       "+
                        "     'table-name' = 'doit42_page_info',               "+
                        "     'zookeeper.quorum' = 'doitedu:2181',             "+
                        "     'lookup.cache' = 'PARTIAL',                      "+
                        "     'lookup.partial-cache.max-rows' = '1000'         "+
                        " )                                                    "
        );


        // 建表，映射kafka中的dwd目标topic
        tenv.executeSql(
                " create table dwd_events_kafka (                       "+
                        "     release_channel string                             "+
                        "     ,device_type string                                 "+
                        "     ,session_id  string                                 "+
                        "     ,lat         double                                 "+
                        "     ,lng         double                                 "+
                        "     ,username    string                                 "+
                        "     ,eventId     string                                 "+
                        "     ,action_time   bigint                                 "+
                        "     ,properties  map<string,string>                     "+
                        "     ,id  bigint                                         "+
                        " 	  ,member_level_id bigint                             "+
                        "     ,password string                                    "+
                        "     ,nickname string                                    "+
                        "     ,phone string                                       "+
                        "     ,status int                                         "+
                        "     ,create_time timestamp(3)                           "+
                        "     ,icon string                                        "+
                        "     ,gender int                                         "+
                        "     ,birthday date                                      "+
                        "     ,register_city string                               "+
                        "     ,job string                                         "+
                        "     ,personalized_signature string                      "+
                        "     ,source_type int                                    "+
                        "     ,integration int                                    "+
                        "     ,growth int                                         "+
                        "     ,luckey_count int                                   "+
                        "     ,history_integration int                            "+
                        "     ,modify_time timestamp(3)                           "+
                        "     ,province string                                    "+
                        "     ,city string                                        "+
                        "     ,region string                                      "+
                        "     ,page_type string                                   "+
                        "     ,page_service string                                "+
                        "     ,page_lanmu string                                  "+
                        "     ,page_pindao string                                 "+
                        " ) with (                                                "+
                        "     'connector' = 'kafka',                              "+
                        "     'topic' = 'dwd-events',                             "+
                        "     'properties.bootstrap.servers' = 'doitedu:9092',    "+
                        "     'value.format'='json',                              "+
                        "     'value.json.fail-on-missing-field'='false',         "+
                        "     'value.fields-include' = 'EXCEPT_KEY'               "+
                        " )                                                       "


        );


        // 建表，映射doris中的dwd目标表
        tenv.executeSql(
                " create table events_detail_doris(                                      "+
                        "    dt  date                                                            "+
                        "    ,member_level_id bigint                                             "+
                        "    ,source_type int                                                    "+
                        "    ,release_channel string                                             "+
                        "    ,device_type   string                                               "+
                        "    ,session_id    varchar(12)                                          "+
                        "    ,lat           double                                               "+
                        "    ,lng           double                                               "+
                        "    ,username      string                                               "+
                        "    ,eventId       string                                               "+
                        "    ,action_time   bigint                                               "+
                        "    ,properties    string                                               "+
                        "    ,user_id        bigint                                               "+
                        "    ,password      string                                               "+
                        "    ,nickname      string                                               "+
                        "    ,phone         string                                               "+
                        "    ,status        int                                                  "+
                        "    ,create_time   timestamp(3)                                         "+
                        "    ,icon          string                                               "+
                        "    ,gender        int                                                  "+
                        "    ,birthday      date                                                 "+
                        "    ,register_city string                                               "+
                        "    ,job           string                                               "+
                        "    ,personalized_signature string                                      "+
                        "    ,integration  int                                                   "+
                        "    ,growth       int                                                   "+
                        "    ,luckey_count int                                                   "+
                        "    ,history_integration int                                            "+
                        "    ,modify_time timestamp(3)                                           "+
                        "    ,province  string                                                   "+
                        "    ,city      string                                                   "+
                        "    ,region    string                                                   "+
                        "    ,page_type string                                                   "+
                        "    ,page_service string                                                "+
                        "    ,page_lanmu   string                                                "+
                        "    ,page_pindao  string                                                "+
                        "    ,primary key (dt,member_level_id,source_type) not enforced          "+
                        " )  with (                                                              "+
                        "     'connector' = 'doris',                                             "+
                        "     'fenodes' = 'doitedu:8030',                                        "+
                        "     'table.identifier' = 'dwd.events_detail',                          "+
                        "     'username' = 'root',                                               "+
                        "     'password' = 'root',                                               "+
                        "     'sink.label-prefix' = 'doit42_no1'                                 "+
                        " )                                                                      "

        );



        // 关联 sql
        tenv.createTemporaryFunction("gps_geo", GpsToGeohashUdf.class);
        tenv.executeSql(
                " create temporary view joined_result AS                                                                                         "+
                        " select   /*+ LOOKUP('table'='user_hbase', 'retry-predicate'='lookup_miss', 'retry-strategy'='fixed_delay', 'fixed-delay'='5s','max-attempts'='3') */      "+
                        "      t1.release_channel                                                                                                        "+
                        "     ,t1.device_type                                                                                                            "+
                        "     ,t1.session_id                                                                                                             "+
                        "     ,t1.lat                                                                                                                    "+
                        "     ,t1.lng                                                                                                                    "+
                        "     ,t1.username                                                                                                               "+
                        "     ,t1.eventId                                                                                                                "+
                        "     ,t1.`timestamp` as action_time                                                                                             "+
                        "     ,t1.properties                                                                                                             "+
                        "     ,t2.f.id as user_id                                                                                                          "+
                        "     ,t2.f.member_level_id                                                                                                        "+
                        "     ,t2.f.password                                                                                                               "+
                        "     ,t2.f.nickname                                                                                                               "+
                        "     ,t2.f.phone                                                                                                                  "+
                        "     ,t2.f.status                                                                                                                 "+
                        "     ,t2.f.create_time                                                                                                            "+
                        "     ,t2.f.icon                                                                                                                   "+
                        "     ,t2.f.gender                                                                                                                 "+
                        "     ,t2.f.birthday                                                                                                               "+
                        "     ,t2.f.city  as register_city                                                                                                 "+
                        "     ,t2.f.job                                                                                                                    "+
                        "     ,t2.f.personalized_signature                                                                                                 "+
                        "     ,t2.f.source_type                                                                                                            "+
                        "     ,t2.f.integration                                                                                                            "+
                        "     ,t2.f.growth                                                                                                                 "+
                        "     ,t2.f.luckey_count                                                                                                           "+
                        "     ,t2.f.history_integration                                                                                                    "+
                        "     ,t2.f.modify_time                                                                                                            "+
                        "     ,t3.f.province                                                                                                               "+
                        "     ,t3.f.city                                                                                                                   "+
                        "     ,t3.f.region                                                                                                                 "+
                        "     ,t4.f.page_type                                                                                                              "+
                        "     ,t4.f.page_service                                                                                                           "+
                        "     ,t4.f.page_lanmu                                                                                                             "+
                        "     ,t4.f.page_pindao                                                                                                            "+
                        " from action_log_kafka t1                                                                                                       "+
                        " LEFT JOIN user_hbase   FOR SYSTEM_TIME AS OF t1.pt   AS t2 ON t1.username=t2.username                                          "+
                        " LEFT JOIN area_hbase   FOR SYSTEM_TIME AS OF t1.pt   AS t3 ON gps_geo(t1.lat,t1.lng)=t3.geohash                                "+
                        " LEFT JOIN page_hbase   FOR SYSTEM_TIME AS OF t1.pt   AS t4 ON regexp_extract(t1.properties['url'],'(^.*/).*?')=t4.url_prefix   "
        );



        // 将关联结果，插入kafka
        tenv.executeSql("insert into dwd_events_kafka select * from joined_result");


        // 将关联结果，插入doris
        // 注册map转json的函数
        tenv.createTemporaryFunction("map_json", MapToJsonStr.class);
        tenv.executeSql(
                " INSERT INTO events_detail_doris                                              "+
                        " SELECT                                                                       "+
                        "   to_date(date_format(to_timestamp_ltz(action_time,3),'yyyy-MM-dd'))  as dt  "+
                        "   ,member_level_id                                                           "+
                        "   ,source_type                                                               "+
                        "   ,release_channel                                                           "+
                        "   ,device_type                                                               "+
                        "   ,session_id                                                                "+
                        "   ,lat                                                                       "+
                        "   ,lng                                                                       "+
                        "   ,username                                                                  "+
                        "   ,eventId                                                                   "+
                        "   ,action_time                                                               "+
                        "   ,map_json(properties)  as properties                                       "+
                        "   ,user_id                                                                        "+
                        "   ,password                                                                  "+
                        "   ,nickname                                                                  "+
                        "   ,phone                                                                     "+
                        "   ,status                                                                    "+
                        "   ,create_time                                                               "+
                        "   ,icon                                                                      "+
                        "   ,gender                                                                    "+
                        "   ,birthday                                                                  "+
                        "   ,register_city                                                             "+
                        "   ,job                                                                       "+
                        "   ,personalized_signature                                                    "+
                        "   ,integration                                                               "+
                        "   ,growth                                                                    "+
                        "   ,luckey_count                                                              "+
                        "   ,history_integration                                                       "+
                        "   ,modify_time                                                               "+
                        "   ,province                                                                  "+
                        "   ,city                                                                      "+
                        "   ,region                                                                    "+
                        "   ,page_type                                                                 "+
                        "   ,page_service                                                              "+
                        "   ,page_lanmu                                                                "+
                        "   ,page_pindao                                                               "+
                        "                                                                              "+
                        " FROM joined_result                                                           "
        );

    }

}
